package cn.com.pc.content.service;

import cn.com.pc.content.domain.content.Content;
import cn.com.pc.content.domain.content.ContentType;
import cn.com.pc.content.dto.*;
import cn.com.pc.content.kafka.consumer.ConsumerException;
import cn.com.pc.content.kafka.producer.ContentProducer;
import cn.com.pc.content.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.web.server.ResponseStatusException;

import java.time.LocalDateTime;
import java.util.Map;

@Slf4j
@Service
public class ContentConsumerService {
    private static final String ADD_ACTION = "create";
    private static final String UPDATE_ACTION = "update";
    private static final String DELETE_ACTION = "delete";
    private static final int KAFKA_KEY_LENGTH = 3;
    private static final int RETRY_TIME = 3;

    @Autowired
    private ContentProducer contentProducer;

    @Autowired
    private ContentService contentService;


    /**
     * 验证物料消息，验证不通过发送问题topic, 生成新物料Id
     *
     * @param messageMap kakfa物料消息
     * @return
     */
    public Object validateMsg(Object key, String message, Map<String, Object> messageMap, ContentType contentType) {

        if (messageMap == null) {
            throw new ConsumerException("kafka消息转化map 为 null");
        }

        String action = String.valueOf(messageMap.get("action"));
        String content = JsonUtils.obj2String(messageMap.get("content"));

        //验证kafka key，验证物料的各种id
        if (!(ADD_ACTION.equals(action)
                || UPDATE_ACTION.equals(action)
                || DELETE_ACTION.equals(action))) {
            throw new ConsumerException("action is invalid!");

        } else if (StringUtils.isEmpty(content)) {
            throw new ConsumerException("kafka msg content is empty!");

        } else if (contentType == null) {
            throw new ConsumerException("kafka key contentType is error!");

        } else {
            //转化为对应内容实体，判断site,biz,app,referId是否为空
            boolean isCorrectContent = true;
            switch (contentType) {
                case Article:
                    ArticleDTO articleDTO = JsonUtils.jsonToObj(content, ArticleDTO.class);
                    isCorrectContent = validContentDTO(articleDTO, messageMap);
                    if (isCorrectContent) {
                        return articleDTO;
                    }
                    break;

                case Video:
                    VideoDTO videoDTO = JsonUtils.jsonToObj(content, VideoDTO.class);
                    isCorrectContent = validContentDTO(videoDTO, messageMap);
                    if (isCorrectContent) {
                        return videoDTO;
                    }
                    break;
                case Post:
                    PostDTO postDTO = JsonUtils.jsonToObj(content, PostDTO.class);
                    isCorrectContent = validContentDTO(postDTO, messageMap);
                    if (isCorrectContent) {
                        return postDTO;
                    }
                    break;
                case Question:
                case Audio:
                case Course:
                case ShortMessage:
                    CustomizedContentDTO customizedContentDTO = JsonUtils.jsonToObj(content, CustomizedContentDTO.class);
                    isCorrectContent = validContentDTO(customizedContentDTO, messageMap);
                    if (isCorrectContent) {
                        return customizedContentDTO;
                    }
                    break;

//                    todo 待增加实体类
//                case AD:
//                    validError = true;
//                    break;
//                case Product:
//                    validError = true;
//                    break;
//                case Other:
//                    validError = true;
//                    break;
                default:
                    throw new ConsumerException("kafka key contentType is error!");
            }
        }
        throw new ConsumerException("kafka 消息验证失败");
    }

    /**
     * 验证kafka 消息 key
     *
     * @param key
     * @return
     */
    public ContentType validateKeyAndGetContentType(String key) {
        ContentType contentType = null;
        try {
            contentType = ContentType.getSourceByName(key.split(":")[0]);
            if (key.split(":").length == KAFKA_KEY_LENGTH
                    && contentType != null) {
                return contentType;
            }
        } catch (Exception e) {
            log.error("validateKeyAndGetContentType exception: " + e);
        }
        if (contentType == null) {
            throw new ConsumerException("kafka key contentType is error!");
        }
        return contentType;
    }

    /**
     * 内容入库
     *
     * @param action      操作动作
     * @param contentDTO  kafkamsg 传入的内容
     * @param content     新对象，无值，用于存储创建的新值或者数据库查出来的值
     * @param contentType 内容类型
     */
    public void handleContentToMongoAndEs(String action, ContentDTO contentDTO, Content content, ContentType contentType) {
        if (contentDTO == null || content == null) {
            return;
        }
        if (StringUtils.isBlank(contentDTO.getId())) {
            contentDTO.setId(null);
        }

        //20210109 增加存在则更新，不存在则创建
        for (int i = 0; i < RETRY_TIME; i++) {
            try {
                if (ADD_ACTION.equals(action)) {
                    Content oldContent = contentService.getContentByReferId(contentDTO.getReferId(), contentDTO.getSite(),
                            contentDTO.getBiz(), contentType);
                    if (oldContent != null) {
                        log.debug("创建时，referId({}),site({}),contentType({})内容已存在，进行更新",
                                contentDTO.getReferId(), contentDTO.getSite(), contentType);
                        updateContent(contentDTO, oldContent, contentType);
                    } else {
                        createContent(contentDTO, content, contentType);
                    }

                } else if (UPDATE_ACTION.equals(action)) {
                    Content oldContent = contentService.getContentByReferId(contentDTO.getReferId(), contentDTO.getSite(),
                            contentDTO.getBiz(), contentType);
                    if (oldContent == null || oldContent.getSite() != contentDTO.getSite()) {
                        log.debug("更新时，referId({}),site({}),contentType({})内容不存在，进行创建",
                                contentDTO.getReferId(), contentDTO.getSite(), contentType);
                        createContent(contentDTO, content, contentType);
                    } else {
                        updateContent(contentDTO, oldContent, contentType);
                    }

                } else if (DELETE_ACTION.equals(action)) {
                    // 判空
                    content = contentService.getContentByReferId(contentDTO.getReferId(), contentDTO.getSite(),
                            contentDTO.getBiz(), contentType);
                    if (content == null || content.getSite() != contentDTO.getSite()) {
                        log.debug("删除时，referId({}),site({}),contentType({})内容不存在",
                                contentDTO.getReferId(), contentDTO.getSite(), contentType);
                        throw new ConsumerException("删除内容时，内容不存在");
                    }

                    contentService.deleteContent(content, contentType);
                    log.debug("delete {}({}) success;", contentType.getName(), content.getId());
                }

                break;
            } catch (DuplicateKeyException e) {
                if (i < RETRY_TIME - 1) {
                    log.info("内容操作异常，唯一索引或主键重复,重试第({})次，referId({})，site({})，contentType({})",
                            i + 1, contentDTO.getReferId(), contentDTO.getSite(), contentType);
                }
                if (i == RETRY_TIME - 1) {
                    log.error("内容操作异常，唯一索引或主键重复,重试后失败; referId({})，site({})，contentType({})",
                            contentDTO.getReferId(), contentDTO.getSite(), contentType);
                    throw e;
                }
            } catch (ResponseStatusException e) {
                if (i < RETRY_TIME - 1) {
                    log.info("内容操作异常，可能查不到mongodb数据导致create时objectId不存在,重试第({})次，referId({})，site({})，contentType({})",
                            i + 1, contentDTO.getReferId(), contentDTO.getSite(), contentType);
                }
                if (i == RETRY_TIME - 1) {
                    log.error("内容操作异常，可能查不到mongodb数据导致create时objectId不存在,重试后失败; referId({})，site({})，contentType({})",
                            contentDTO.getReferId(), contentDTO.getSite(), contentType);
                    throw e;
                }
            }
        }
    }

    /**
     * 新建内容
     *
     * @param contentDTO
     * @param content
     * @param contentType
     */
    public void createContent(ContentDTO contentDTO, Content content, ContentType contentType) {
        contentDTO.setCreatedAt(contentDTO.getCreatedAt() == null
                ? LocalDateTime.now() : contentDTO.getCreatedAt());
        contentDTO.setUpdatedAt(contentDTO.getCreatedAt());
        contentDTO.setCreatedBy(contentDTO.getCreatedBy() == null ? "anonymous" : contentDTO.getCreatedBy());
        contentDTO.setUpdatedBy(contentDTO.getCreatedBy());
        //创建文章时，id设为null
        //contentDTO.setId(null);
        String contentId = contentService.createContent(contentDTO, content, contentType).getId();
        log.debug("create {}({}) success;", contentType.getName(), contentId);
    }

    /**
     * 更新内容
     *
     * @param contentDTO
     * @param content
     * @param contentType
     */
    public void updateContent(ContentDTO contentDTO, Content content, ContentType contentType) {
        if (content == null || content.getSite() != contentDTO.getSite()) {
            throw new ConsumerException("更新内容时，内容不存在");
        }
        contentDTO.setUpdatedAt(contentDTO.getUpdatedAt() == null
                ? LocalDateTime.now() : contentDTO.getUpdatedAt());
        contentDTO.setUpdatedBy(contentDTO.getUpdatedBy() == null ? "anonymous" : contentDTO.getUpdatedBy());
        contentService.updateContent(contentDTO, content, contentType);
        log.debug("update {}({}) success;", contentType.getName(), contentDTO.getId());
    }

    /**
     * 判断内容的site,biz,app,referId,title是否为空
     *
     * @param contentDTO
     * @param messageMap
     * @return
     */
    public boolean validContentDTO(ContentDTO contentDTO, Map<String, Object> messageMap) {
        if (contentDTO == null) {
            throw new ConsumerException("content parse to entity error or miss necessary field!");
        } else if (contentDTO.getSite() == null
                || StringUtils.isEmpty(contentDTO.getReferId())
//                || StringUtils.isEmpty(contentDTO.getTitle())
                || contentDTO.getBiz() == null
                || contentDTO.getApp() == null) {
            throw new ConsumerException("site/biz/app/referId is null or empty!");
        }
        return true;
    }
}
